package prictise.com.application1.rxjava;

import android.app.Activity;
import android.os.Bundle;
import android.os.Handler;
import android.os.Looper;
import android.os.Message;
import android.support.annotation.Nullable;

import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;

import java.util.Arrays;
import java.util.List;
import java.util.concurrent.TimeUnit;

import io.reactivex.Observable;
import io.reactivex.ObservableSource;
import io.reactivex.Observer;
import io.reactivex.Single;
import io.reactivex.SingleEmitter;
import io.reactivex.SingleObserver;
import io.reactivex.SingleOnSubscribe;
import io.reactivex.android.schedulers.AndroidSchedulers;
import io.reactivex.annotations.NonNull;
import io.reactivex.disposables.Disposable;
import io.reactivex.functions.Consumer;
import io.reactivex.functions.Function;
import io.reactivex.schedulers.Schedulers;
import prictise.com.application1.R;
import prictise.com.application1.utils.LogcatUtils;

public class RxJavaActivity1 extends Activity {
  private String TAG = "RxJavaActivity1";
  
  Handler handler = new Handler() {
    @Override
    public void handleMessage(Message msg) {
    
    }
  };
  @Override
  protected void onCreate(@Nullable Bundle savedInstanceState) {
    super.onCreate(savedInstanceState);
    setContentView(R.layout.activity_rxjava);
    
    new Thread(new Runnable() {
      @Override
      public void run() {
        
        
        handler.sendEmptyMessage(1);
      }
    }).start();
    
    
    Observable.fromArray(1, 2, 3, 4, 5).flatMap(new Function<Integer, ObservableSource<Integer>>() {
      @Override
      public ObservableSource<Integer> apply(@NonNull Integer integer) throws Exception {
        int delay = 0;
        if (integer == 3) {
          delay = 1500;
        }
        
        return Observable.just(integer * 10).delay(delay, TimeUnit.MILLISECONDS);
      }
    })
      .subscribeOn(Schedulers.io())
      .observeOn(AndroidSchedulers.mainThread())
      .subscribe(new Consumer<Integer>() {
        @Override
        public void accept(Integer integer) throws Exception {
          LogcatUtils.showDLog(TAG, "subscribe integer = " + integer);
        }
      });
    
//    singleTest();
//    singleCreate();
    singleFlatMap();
  }
  
  private void singleFlatMap() {
    query().flatMap(new Function<List<String>, ObservableSource<String>>() {
      @Override
      public ObservableSource<String> apply(@NonNull List<String> strings) throws Exception {
        return Observable.fromIterable(strings);
      }
    }).subscribe(new Observer<String>() {
      @Override
      public void onSubscribe(@NonNull Disposable d) {
        LogcatUtils.showDLog(TAG, "singleFlatMap onSubscribe");
      }
  
      @Override
      public void onNext(@NonNull String s) {
        LogcatUtils.showDLog(TAG, "singleFlatMap onNext s = " + s);
      }
  
      @Override
      public void onError(@NonNull Throwable e) {
        LogcatUtils.showDLog(TAG, "singleFlatMap onError");
      }
  
      @Override
      public void onComplete() {
        LogcatUtils.showDLog(TAG, "singleFlatMap onComplete");
      }
    });
  }
  
  Observable<List<String>> query() {
    List<String> s = Arrays.asList("Java", "Android", "Ruby", "Ios", "Swift");
    return Observable.just(s);
  }
  
  private void singleCreate() {
    Single.create(new SingleOnSubscribe<Integer>() {
      @Override
      public void subscribe(@NonNull SingleEmitter<Integer> emitter) throws Exception {
        LogcatUtils.showDLog(TAG, "subscribe");
        emitter.onSuccess(addValue(2, 3));
      }
    }).subscribeOn(Schedulers.io()).subscribe(new SingleObserver<Integer>() {
      
      @Override
      public void onSubscribe(@NonNull Disposable d) {
        LogcatUtils.showDLog(TAG, "onSubscribe");
      }
      
      @Override
      public void onSuccess(@NonNull Integer integer) {
        LogcatUtils.showDLog(TAG, "onSuccess integer = " + integer);
      }
      
      @Override
      public void onError(@NonNull Throwable e) {
        LogcatUtils.showDLog(TAG, "onError");
      }
    });
    // 常见的示例，这是一个异步操作
//    Single.create(new SingleOnSubscribe<Integer>() {
//      @Override
//      public void call(SingleSubscriber<? super Integer> singleSubscriber) {
//        // 这里被指定在IO线程
//        singleSubscriber.onSuccess(addValue(1, 2));
//      }
//    }).subscribeOn(Schedulers.io())// 指定运行在IO线程
//      .subscribe(new Subscriber<Integer>() {
//        @Override
//        public void onCompleted() {            }
//        @Override
//        public void onError(Throwable e) {    }
//        @Override
//        public void onNext(Integer o) {
//          // o = 3
//        }
//      });
  }
  
  private void singleTest() {
    Single.just(addValue(3, 5)).subscribe(new SingleObserver<Integer>() {
      @Override
      public void onSubscribe(@NonNull Disposable d) {
      
      }
      
      @Override
      public void onSuccess(@NonNull Integer integer) {
        LogcatUtils.showDLog(TAG, "integer = " + integer);
      }
      
      @Override
      public void onError(@NonNull Throwable e) {
      
      }
    });
  }
  
  int addValue(int a, int b) {
    return a + b;
  }
}
